package com.pw.study.handler;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import com.pw.study.common.constants.TopicConstant;

import java.util.List;
import java.util.Random;

public class MessageHandler {

    public static void doDataDeal(Message message) {

        System.out.println("开始处理数据。。。。");
        List<CanalEntry.Entry> entries = message.getEntries();

        for (CanalEntry.Entry entry : entries) {
            System.out.println(entry.getEntryType());
            //只处理增删改数据
            if(entry.getEntryType().equals(CanalEntry.EntryType.ROWDATA)){
                //从header中取出表数据
                String tableName = entry.getHeader().getTableName();
                System.out.println("tableName:"+ tableName);

                //获取数据需要反序列化
                ByteString storeValue = entry.getStoreValue();
                parseStoreValue(storeValue);

            }
        }

        System.out.println("结束处理数据。。。。");
    }
    public static void doDataDealByOrder(Message message) {

        System.out.println("开始处理数据。。。。");
        List<CanalEntry.Entry> entries = message.getEntries();

        for (CanalEntry.Entry entry : entries) {
            System.out.println(entry.getEntryType());
            //只处理增删改数据
            if(entry.getEntryType().equals(CanalEntry.EntryType.ROWDATA)){
                //从header中取出表数据
                String tableName = entry.getHeader().getTableName();
                ByteString storeValue = entry.getStoreValue();
                System.out.println("tableName:"+ tableName);
                if(tableName.equals("order_detail")){
                    //获取数据需要反序列化
                    sendDataToKafka(TopicConstant.GMALL_ORDER_DETAIL,storeValue);
                }
                if(tableName.equals("order_info")){
                    sendDataToKafka(TopicConstant.GMALL_ORDER_INFO,storeValue);
                }
                if(tableName.equals("user_info")){
                    sendDataToKafka(TopicConstant.GMALL_USER_INFO,storeValue);
                }

            }
        }

        System.out.println("结束处理数据。。。。");
    }
    private static void sendDataToKafka(String topic, ByteString storeValue) {
        System.out.println("数据反序列化。。。。。。");
        try {
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
            CanalEntry.EventType eventType = rowChange.getEventType();
            //获取insert数据
            if(eventType== CanalEntry.EventType.INSERT){
                List<CanalEntry.RowData> rowList = rowChange.getRowDatasList();
                for (CanalEntry.RowData rowData : rowList) {
                    //每行数据封装
                    JSONObject obj = new JSONObject();
                    List<CanalEntry.Column> columnsList = rowData.getAfterColumnsList();
                    for (CanalEntry.Column column : columnsList) {
                        System.out.println("obj.." +column.toString());
                        obj.put(column.getName(),column.getValue());
                    }
                    //数据写入kafka
                    KafkaHandler.sendDataToKafka(topic,obj.toString());
                }
            }


        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
    }
    private static void parseStoreValue(ByteString storeValue) {
        System.out.println("数据反序列化。。。。。。");
        try {
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
            CanalEntry.EventType eventType = rowChange.getEventType();
            //获取insert数据
          if(eventType== CanalEntry.EventType.INSERT){
              List<CanalEntry.RowData> rowList = rowChange.getRowDatasList();
              for (CanalEntry.RowData rowData : rowList) {
                  //每行数据封装
                  JSONObject obj = new JSONObject();
                  List<CanalEntry.Column> columnsList = rowData.getAfterColumnsList();
                  for (CanalEntry.Column column : columnsList) {
                      System.out.println("obj.." +column.toString());
                      obj.put(column.getName(),column.getValue());
                  }
                  //数据写入kafka
                  KafkaHandler.sendDataToKafka(TopicConstant.GMALL_ORDER_INFO,obj.toString());
              }
          }


        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
    }
}
